package com.winit.schedule.task.tracker.service;

import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.winit.schedule.core.conf.Configuration;
import com.winit.schedule.core.exception.LocalStoreException;
import com.winit.schedule.core.localstore.KVPair;
import com.winit.schedule.core.localstore.LocalStore;
import com.winit.schedule.core.localstore.LocalStoreConfig;
import com.winit.schedule.core.localstore.LocalStoreFactory;
import com.winit.schedule.core.localstore.leveldb.LeveldbLocalStoreFactory;
import com.winit.schedule.core.protocol.command.CompleteTaskRequest;
import com.winit.schedule.core.repository.entity.TaskTrackerEntity;
import com.winit.schedule.core.service.AbstractService;
import com.winit.schedule.core.utils.Constants;
import com.winit.schedule.core.utils.Md5Encrypt;
import com.winit.schedule.task.tracker.TaskTrackerContext;
import com.winit.schedule.task.tracker.utils.RemoteUtil;

public class RPCRetryService extends AbstractService {

    private static transient final Logger  logger            = LoggerFactory.getLogger(RPCRetryService.class);

    /**
     * rty任务重试定时任务
     */
    private final ScheduledExecutorService retryExecutor     = Executors.newScheduledThreadPool(1);

    private LocalStoreFactory              localStoreFactory = new LeveldbLocalStoreFactory();

    private TaskTrackerContext             taskTrackerContext;

    private LocalStore                     localStore;

    public RPCRetryService(TaskTrackerContext taskTrackerContext){
        super("RPCRetryService");
        this.taskTrackerContext = taskTrackerContext;
    }

    @Override
    protected void serviceInit(Configuration conf) throws Exception {
        super.serviceInit(conf);

        TaskTrackerEntity taskTracker = taskTrackerContext.getTaskTracker().getTaskTrackerEntity();
        
        // 初始化本地存储
        LocalStoreConfig config = new LocalStoreConfig();
        config.setLocalStorePath(Constants.USER_HOME);
        config.setTaskGroup(taskTracker.getTaskGroup());
        String identity = taskTracker.getTaskTrackerUuid();
        config.setIdentity(identity);
        localStore = localStoreFactory.getLocalStore(config, null);
        localStore.open();
    }

    @Override
    protected void serviceStart() throws Exception {
        super.serviceStart();
        retryExecutor.scheduleWithFixedDelay(new RetryThread(this), 3000, 5000, TimeUnit.MILLISECONDS);
    }

    @Override
    protected void serviceStop() throws Exception {
        super.serviceStop();
        retryExecutor.shutdown();
    }

    public TaskTrackerContext getTaskTrackerContext() {
        return taskTrackerContext;
    }

    public void addRetryTask(CompleteTaskRequest task, int maxRetryTimes) {
        if (null != task) {
            logger.info("add retry task : " + task + " , maxRetryTimes : " + maxRetryTimes);
            try {
                localStore.put(task.getIdentity(), task);
            } catch (LocalStoreException e) {
                logger.error("add retry task failure.", e);
            }
        }
    }

    class RetryThread implements Runnable {

        RPCRetryService retryService;

        public RetryThread(RPCRetryService retryService){
            this.retryService = retryService;
        }

        @Override
        public void run() {
            try {
                Thread.currentThread().setPriority(Thread.MAX_PRIORITY);

                List<KVPair<String, CompleteTaskRequest>> datas = localStore.fetchTop(100, CompleteTaskRequest.class);

                if (null != datas && datas.size() > 0) {
                    for (KVPair<String, CompleteTaskRequest> pair : datas) {
                        CompleteTaskRequest commandBody = pair.getValue();
                        if (null != commandBody) {

                            boolean success = RemoteUtil.sendTaskCompleteCommand(commandBody,
                                retryService.getTaskTrackerContext().getRemotingClientService());

                            // 发送失败，重新加入队列
                            if (!success) {
                                if (logger.isDebugEnabled()) {
                                    logger.debug("Retry sendTaskCompleteCommand failure.");
                                }
                            } else {
                                localStore.delete(pair.getKey());
                            }
                        } else {
                            localStore.delete(pair.getKey());
                        }
                    }
                }
            } catch (Exception ex) {
                logger.error("Retry sendTaskCompleteCommand failure.", ex);
            }
        }

    }

}
